package com.shlh.saas.service.twitter.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.shlh.saas.common.Desensitization;
import com.shlh.saas.entity.company.Company;
import com.shlh.saas.entity.facebook.FacebookUser;
import com.shlh.saas.entity.facebook.FacebookUserEs;
import com.shlh.saas.entity.twitter.TwitterUser;
import com.shlh.saas.entity.twitter.TwitterUserEs;
import com.shlh.saas.mapper.twitter.TwitterMapper;
import com.shlh.saas.model.dto.TwitterQueryDTO;
import com.shlh.saas.service.twitter.TwitterEsService;
import com.shlh.saas.service.twitter.TwitterService;
import com.shlh.saas.util.ExcelUtils;
import com.shlh.saas.util.TwitterUserDataSource;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.formula.functions.T;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Twitter用户服务实现类
 */
@Service
@Slf4j
public class TwitterServiceImpl extends ServiceImpl<TwitterMapper, TwitterUser> implements TwitterService {

    @Autowired
    private TwitterMapper twitterMapper;

    @Resource
    private TwitterEsService twitterEsService;

    @Resource
    private JdbcTemplate jdbcTemplate;

    @Resource
    private TwitterUserDataSource twitterUserDataSource;

    @Autowired
    private Desensitization desensitization;

    /**
     * 根据邮箱查询Twitter用户
     * @param email 邮箱
     * @return Twitter用户
     */
    @Override
    public List<TwitterUser> findByEmail(String email) {
        List<Map<String, String>> maps = twitterEsService.searchByEmail(email, 10, 0);

        List<TwitterUser> twitterUserList = new ArrayList<>();

        for (Map<String, String> map : maps) {
            String mysqlTable = map.get("sourceTable");
            String id = map.get("sourceId");
            TwitterUser twitterUser = getTwitterUserByTableAndId(mysqlTable, id);
            twitterUserList.add(twitterUser);
        }

        return twitterUserList;
    }

    private TwitterUser getTwitterUserByTableAndId(String mysqlTable, String id) {
        log.info("根据ID查询企业: {}", id);
        try {
            log.info("根据数据表和ID查询企业: mysqlTable={}, id={}", mysqlTable, id);

            // 参数验证
            if (!StringUtils.hasText(mysqlTable) || !StringUtils.hasText(id)) {
                log.warn("参数不能为空: mysqlTable={}, id={}", mysqlTable, id);
                return null;
            }

            // 验证数据表名是否安全（防止SQL注入）
            if (!isValidTableName(mysqlTable)) {
                log.warn("无效的数据表名: {}", mysqlTable);
                return null;
            }
            // 查询twitter_user信息

            // 使用ShardingSphere兼容的查询方式
            List<Map<String, Object>> results = executeShardingSphereTwitterQuery(mysqlTable, id);

            if (results.isEmpty()) {
                log.info("未查询到企业信息: table={}, id={}",mysqlTable, id);
                return null;
            }

            // 将查询结果转换为TwitterUser对象
            Map<String, Object> row = results.get(0);
            TwitterUser twitterUser = convertMapToTwitterUser(row);

            log.info("成功获取联系人详情: {}",
                    JSON.toJSONString(twitterUser));

            return twitterUser;


        } catch (Exception e) {
            log.error("根据ID查询企业失败", e);
            return null;
        }
    }

    private TwitterUser convertMapToTwitterUser(Map<String, Object> row) {
        TwitterUser twitterUser = new TwitterUser();

        twitterUser.setId((Long) row.get("id"));
        twitterUser.setName(getString(row, "name"));
        twitterUser.setEmail(getString(row, "email"));
        twitterUser.setScreenName(getString(row, "screen_name"));
        twitterUser.setFollowers((Integer) row.get("followers"));

        return twitterUser;
    }

    /**
     * 安全地从Map中获取字符串值
     *
     * @param row 数据行
     * @param key 键名
     * @return 字符串值
     */
    private String getString(Map<String, Object> row, String key) {
        Object value = row.get(key);
        return value != null ? value.toString() : null;
    }

    private List<Map<String, Object>> executeShardingSphereTwitterQuery(String mysqlTable, String id) {

        try {
            // 使用ShardingSphere兼容的查询方式
            jdbcTemplate.setDataSource(twitterUserDataSource.dataSource());

            // 使用ShardingSphere Hint强制指定数据源
            String sql =
                    "/* ShardingSphere hint: dataSourceName=twitter_db */ SELECT * FROM twitter_db." + mysqlTable + " WHERE id = ?";
            log.debug("执行备用查询: {}, 参数: {}", sql, id);

            return jdbcTemplate.queryForList(sql, id);

        } catch (Exception e) {
            log.error("执行备用查询失败", e);
            return new ArrayList<>();
        }
    }

    private boolean isValidTableName(String mysqlTable) {
        // 识别表名twitter_user_0到twitter_user_199
        return mysqlTable.matches("^twitter_user_(?:0|[1-9]\\d?|1[0-9]{2})$");
    }


    /**
     * 根据推特账号查询Twitter用户
     * @param screenName 推特账号
     * @return Twitter用户
     */
    @Override
    public TwitterUser findByScreenName(String screenName) {
        LambdaQueryWrapper<TwitterUser> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(TwitterUser::getScreenName, screenName);
        return getOne(wrapper);
    }
    
    @Override
    public Page<TwitterUser> getTwitterUserPage(TwitterQueryDTO queryDTO) {

        TwitterUserEs twitterUserEs = covertToTwitterUserEs(queryDTO);

        Page<TwitterUserEs> twitterUserEsPage = twitterEsService.queryTwitterUserList(twitterUserEs);

        Page<TwitterUser> twitterUserPage =  covertToTwitterUserPage(twitterUserEsPage);
        return twitterUserPage;
    }

    private Page<TwitterUser> covertToTwitterUserPage(Page<TwitterUserEs> twitterUserEsPage) {
        Page<TwitterUser> twitterUserPage = new Page<>();

        twitterUserPage.setCurrent(twitterUserEsPage.getCurrent());
        twitterUserPage.setSize(twitterUserEsPage.getSize());
        List<TwitterUser> facebookUserList = new LinkedList<>();
        for (TwitterUserEs twitterUserEs : twitterUserEsPage.getRecords()) {
            TwitterUser TwitterUser;
            TwitterUser = covertToTwitterUser(twitterUserEs);
            facebookUserList.add(TwitterUser);
        }
        twitterUserPage.setRecords(facebookUserList);
        twitterUserPage.setTotal(twitterUserEsPage.getTotal());
        return twitterUserPage;
    }

    private Page<TwitterUser> covertToTwitterUserPageEncryption(Page<TwitterUserEs> twitterUserEsPage) {
        Page<TwitterUser> twitterUserPage = new Page<>();

        twitterUserPage.setCurrent(twitterUserEsPage.getCurrent());
        twitterUserPage.setSize(twitterUserEsPage.getSize());
        List<TwitterUser> facebookUserList = new LinkedList<>();
        for (TwitterUserEs twitterUserEs : twitterUserEsPage.getRecords()) {
            TwitterUser TwitterUser;
            TwitterUser = covertToTwitterUserEncryption(twitterUserEs);
            facebookUserList.add(TwitterUser);
        }
        twitterUserPage.setRecords(facebookUserList);
        twitterUserPage.setTotal(twitterUserEsPage.getTotal());
        return twitterUserPage;
    }

    private TwitterUser covertToTwitterUser(TwitterUserEs twitterUserEs) {
        TwitterUser twitterUser = new TwitterUser();

        if (twitterUserEs.getSourceId() != null) {
            try {
                twitterUser.setId(Long.parseLong(twitterUserEs.getSourceId()));
            } catch (NumberFormatException e) {
                log.warn("无法转换sourceId为Long类型: {}", twitterUserEs.getSourceId());
            }
        }

        // 基本信息
        twitterUser.setName(twitterUserEs.getName());
        twitterUser.setEmail(twitterUserEs.getEmail());
        twitterUser.setScreenName(twitterUserEs.getScreenName());
        twitterUser.setCreatedAt(twitterUserEs.getCreatedAt());

        // 粉丝数相关字段
        twitterUser.setFollowers(twitterUserEs.getFollowers());
        twitterUser.setFollowersCount(twitterUserEs.getFollowers()); // 用于导出兼容性

        // 设置其他导出需要的字段（这些字段在ES中可能不存在，设置默认值）
        twitterUser.setVerified(twitterUserEs.getVerified());

        // 注意：以下字段在当前ES结构中不存在，设置为null或默认值
        // 如果需要这些字段的真实数据，需要从其他数据源获取或扩展ES索引
//        twitterUser.setFriendsCount(null);
//        twitterUser.setStatusesCount(null);
//        twitterUser.setListedCount(null);
        twitterUser.setGender(null);
        twitterUser.setUrl(null);
        twitterUser.setDescription(null);
        twitterUser.setLastStatusSource(null);
        twitterUser.setPhone(null);
        twitterUser.setLocation(null);

        return twitterUser;
    }

    private TwitterUser covertToTwitterUserEncryption(TwitterUserEs twitterUserEs) {
        TwitterUser twitterUser = new TwitterUser();

        if (twitterUserEs.getSourceId() != null) {
            try {
                twitterUser.setId(Long.parseLong(twitterUserEs.getSourceId()));
            } catch (NumberFormatException e) {
                log.warn("无法转换sourceId为Long类型: {}", twitterUserEs.getSourceId());
            }
        }

        // 基本信息
        twitterUser.setName(twitterUserEs.getName());
        twitterUser.setEmail(desensitization.maskEmail(twitterUserEs.getEmail()));
        twitterUser.setScreenName(twitterUserEs.getScreenName());
        twitterUser.setCreatedAt(twitterUserEs.getCreatedAt());

        // 粉丝数相关字段
        twitterUser.setFollowers(twitterUserEs.getFollowers());
        twitterUser.setFollowersCount(twitterUserEs.getFollowers()); // 用于导出兼容性

        // 设置其他导出需要的字段（这些字段在ES中可能不存在，设置默认值）
        twitterUser.setVerified(twitterUserEs.getVerified());

        // 数据库字段
        twitterUser.setSourceTable(twitterUserEs.getSourceTable());
        twitterUser.setSourceId(twitterUserEs.getSourceId());

        // 注意：以下字段在当前ES结构中不存在，设置为null或默认值
        // 如果需要这些字段的真实数据，需要从其他数据源获取或扩展ES索引
//        twitterUser.setFriendsCount(null);
//        twitterUser.setStatusesCount(null);
//        twitterUser.setListedCount(null);
        twitterUser.setGender(null);
        twitterUser.setUrl(null);
        twitterUser.setDescription(null);
        twitterUser.setLastStatusSource(null);
        twitterUser.setPhone(null);
        twitterUser.setLocation(null);

        return twitterUser;
    }

    private TwitterUserEs covertToTwitterUserEs(TwitterQueryDTO queryDTO) {
        TwitterUserEs twitterUserEs = new TwitterUserEs();

        twitterUserEs.setCurrent(queryDTO.getPageNum());
        twitterUserEs.setSize(queryDTO.getPageSize());

        if (queryDTO.getEmail() != null && !queryDTO.getEmail().isEmpty()) {
            twitterUserEs.setEmail(queryDTO.getEmail());
        }

        if (queryDTO.getFollowers_range() != null && queryDTO.getFollowers_range().length == 2) {
            twitterUserEs.setMinFollowers(queryDTO.getFollowers_range()[0]);

            twitterUserEs.setMaxFollowers(queryDTO.getFollowers_range()[1]);
        }

        if (queryDTO.getName() != null && !queryDTO.getName().isEmpty()) {
            twitterUserEs.setName(queryDTO.getName());
        }

        if (queryDTO.getScreenName() != null && !queryDTO.getScreenName().isEmpty()){
            twitterUserEs.setScreenName(queryDTO.getScreenName());
        }

        if(queryDTO.getCreated_at_year() != null && !queryDTO.getCreated_at_year().isEmpty()){
            twitterUserEs.setCreatedYear(queryDTO.getCreated_at_year());
        }

        return twitterUserEs;
    }

    @Override
    public List<TwitterUser> exportTwitterUsers(TwitterQueryDTO queryDTO) {
        log.info("导出Twitter用户数据，查询参数：{}", queryDTO);
        List<TwitterUser> twitterUserList = new ArrayList<>();
        
        try {
            // 创建TwitterUserEs查询对象
            TwitterUserEs twitterUserEs = covertToTwitterUserEs(queryDTO);
            
            // 限制导出数量
            if (queryDTO.getLimit() != null && queryDTO.getLimit() > 0) {
                twitterUserEs.setSize(queryDTO.getLimit());
                log.info("使用指定导出数量限制: {}", queryDTO.getLimit());
            } else {
                twitterUserEs.setSize(50000);
                log.info("使用默认导出数量限制: 50000");
            }
            
            // 获取总数量
            long totalCount = twitterEsService.countTwitterUsersWithKeyWord(queryDTO.getName());
            log.info("符合条件的总记录数：{}", totalCount);
            
            // 根据数量决定查询策略
            int limit = Math.toIntExact(twitterUserEs.getSize());
            if (limit > 10000) {
                log.info("数据量大于10000，使用分批查询策略");
                
                // 使用分批次查询
                int batchSize = 5000; // 每批5000条
                int totalBatches = (int) Math.ceil((double) limit / batchSize);
                log.info("总共需要查询{}批，每批{}条", totalBatches, batchSize);
                
                for (int i = 0; i < totalBatches; i++) {
                    int from = i * batchSize;
                    // 确保最后一批不超过总限制
                    int currentBatchSize = Math.min(batchSize, limit - from);
                    if (currentBatchSize <= 0) break; // 安全检查
                    
                    log.info("正在查询第{}批数据，起始位置：{}，批量大小：{}", i + 1, from, currentBatchSize);
                    
                    // 分批查询ES数据
                    TwitterUserEs batchQuery = new TwitterUserEs();
                    batchQuery.setName(twitterUserEs.getName());
                    batchQuery.setCurrent(from / currentBatchSize + 1); // 页码从1开始
                    batchQuery.setSize(currentBatchSize);
                    
                    Page<TwitterUserEs> batchResult = twitterEsService.queryTwitterUserList(batchQuery);
                    List<TwitterUser> batchList = covertToTwitterUserPage(batchResult).getRecords();
                    
                    log.info("第{}批数据查询完成，获取到{}条数据", i + 1, batchList.size());
                    twitterUserList.addAll(batchList);
                    
                    // 简单进度报告
                    log.info("已完成 {}/{}，当前累计获取数据量: {}", 
                           i + 1, totalBatches, twitterUserList.size());
                }
            } else {
                // 对于小数据量，使用原来的查询方法
                log.info("数据量较小，使用常规查询策略");
                
                // 导出时从第一条开始
                twitterUserEs.setCurrent(1);
                Page<TwitterUserEs> twitterUserEsPage = twitterEsService.queryTwitterUserList(twitterUserEs);
                twitterUserList = covertToTwitterUserPage(twitterUserEsPage).getRecords();
            }
            
            log.info("导出查询完成，返回{}条数据", twitterUserList.size());
        } catch (Exception e) {
            log.error("导出Twitter用户数据查询失败", e);
            // 返回空列表而不是抛出异常，让调用方能够处理
            return new ArrayList<>();
        }
        
        return twitterUserList;
    }
    
    @Override
    public TwitterUser getTwitterUserById(Integer id) {
        return getById(id);
    }
    
    @Override
    public Page<TwitterUser> getTwitterUsersPage(TwitterQueryDTO queryDTO) {
        Page<TwitterUser> twitterUserPage;

        TwitterUserEs twitterUserEs = covertToTwitterUserEs(queryDTO);

        Page<TwitterUserEs> twitterUserEsPage = twitterEsService.queryTwitterUserList(twitterUserEs);

        twitterUserPage = covertToTwitterUserPageEncryption(twitterUserEsPage);

        return twitterUserPage;

    }
    
    @Override
    public List<TwitterUser> getTwitterUsersList(TwitterQueryDTO queryDTO) {
        // 调用导出方法，复用查询逻辑
        return exportTwitterUsers(queryDTO);
    }

    /**
     * 导入Twitter用户数据
     *
     * @param file Excel文件
     * @return 导入的Twitter用户列表
     * @throws Exception 导入异常
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public List<TwitterUser> importTwitterUsers(MultipartFile file) throws Exception {
        if (file == null) {
            throw new IllegalArgumentException("文件不能为空");
        }
        
        try {
            // 使用ExcelUtils工具类导入Excel
            List<TwitterUser> importList = ExcelUtils.importExcel(file, TwitterUser.class);
            
            if (importList.isEmpty()) {
                return importList;
            }
            
            // 过滤无效数据
            importList = importList.stream()
                    .filter(data -> data.getId() != null)
                    .collect(Collectors.toList());
            
            if (importList.isEmpty()) {
                return importList;
            }
            
            // 批量保存或更新
            saveOrUpdateBatch(importList);
            
            return importList;
        } catch (IOException e) {
            throw new Exception("读取Excel文件失败", e);
        } catch (Exception e) {
            throw new Exception("导入数据失败: " + e.getMessage(), e);
        }
    }
    
    @Override
    public List<Map<String, Object>> searchByKeywordInAllFields(String keyword, int offset, int limit) {
        log.info("在Twitter用户所有字段中搜索关键词: {}, 偏移量: {}, 限制数: {}", keyword, offset, limit);
        try {

            List<Map<String, String>> maps = twitterEsService.searchByKeyWord(keyword, offset, limit);

            List<Map<String, Object>> results = new ArrayList<>();
            for (Map<String, String> map : maps){
                String mysqlTable = map.get("sourceTable");
                String id = map.get("sourceId");
                Map<String, Object> twitterUserMap = getTwitterUserMapByTableAndId(mysqlTable, id);
                results.add(twitterUserMap);
            }
            return results;

        } catch (Exception e) {
            log.error("Twitter用户全字段搜索失败", e);
            throw e;
        }
    }
    
    @Override
    public List<Map<String, String>> searchByKeyWordForIdentifiers(String keyword, int offset, int limit) {
        log.info("从ES查询Twitter标识信息: keyword={}, offset={}, limit={}", keyword, offset, limit);
        try {
            return twitterEsService.searchByKeyWord(keyword, offset, limit);
        } catch (Exception e) {
            log.error("查询Twitter标识信息失败", e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Map<String, Object>> searchByKeyword(String keyword, int offset, int limit) {
        log.info("从ES查询Twitter标识信息: keyword={}, offset={}, limit={}", keyword, offset, limit);
        try {
            return twitterEsService.findByKeyWord(keyword, offset, limit);
        } catch (Exception e) {
            log.error("查询Twitter标识信息失败", e);
            return new ArrayList<>();
        }
    }
    
    @Override
    public List<Map<String, Object>> searchByNameAndUsernameForIdentifiers(String name, String username, int limit) {
        log.info("从ES查询Twitter标识信息: name={}, username={}, limit={}", name, username, limit);
        try {
            return twitterEsService.searchByNameAndScreenName(name, username, limit);
        } catch (Exception e) {
            log.error("查询Twitter标识信息失败: name={}, username={}", name, username, e);
            return new ArrayList<>();
        }
    }

    @Override
    public long countByKeywordInAllFields(String keyword) {
        log.info("统计Twitter用户中包含关键词: {} 的记录数", keyword);
        try {
            // 统计在所有字段中包含关键词的记录总数
            return twitterEsService.countTwitterUsersWithKeyWord(keyword);
        } catch (Exception e) {
            log.error("统计Twitter用户全字段搜索结果数量失败", e);
            throw e;
        }
    }
    
    @Override
    public List<String> getSuggestions(String keyword) {
        log.info("获取Twitter用户搜索建议，关键词: {}", keyword);
        if (!StringUtils.hasText(keyword) || keyword.length() < 2) {
            return new ArrayList<>();
        }
        
        try {
            // 从用户名中查找
            List<String> usernames = twitterMapper.findSimilarUsernames(keyword);
            
            // 从显示名称中查找
            List<String> displayNames = twitterMapper.findSimilarDisplayNames(keyword);
            
            // 从个人简介中查找
            List<String> bios = twitterMapper.findSimilarBios(keyword);
            
            // 合并结果并去重
            List<String> suggestions = new ArrayList<>();
            suggestions.addAll(usernames);
            suggestions.addAll(displayNames);
            suggestions.addAll(bios);
            
            // 去重并限制数量
            return suggestions.stream()
                    .distinct()
                    .limit(10)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            log.error("获取Twitter用户搜索建议失败", e);
            return new ArrayList<>();
        }
    }
    
    @Override
    public List<TwitterUser> getAllTwitterUsers() {
        try {
            log.info("正在查询所有Twitter用户数据");
            List<TwitterUser> users = twitterMapper.selectList(null);
            log.info("查询Twitter用户数据完成，找到{}条记录", users.size());
            if (users.isEmpty()) {
                log.warn("数据库中没有Twitter用户数据");
            } else {
                log.info("Twitter用户数据示例: {}", users.get(0));
            }
            return users;
        } catch (Exception e) {
            log.error("查询Twitter用户数据失败: {}", e.getMessage(), e);
            return new ArrayList<>();
        }
    }
    
    @Override
    public long countTwitterUsersByKeyword(String keyword) {
        try {
            log.info("统计Twitter用户数据，关键词: {}", keyword);

            long count = countByKeywordInAllFields(keyword);
            log.info("统计Twitter用户数据完成，关键词: {}, 找到{}条记录", keyword, count);
            return count;
        } catch (Exception e) {
            log.error("统计Twitter用户数据失败，关键词: {}, 错误: {}", keyword, e.getMessage(), e);
            return 0;
        }
    }
    
    @Override
    public long countTwitterUsers(TwitterQueryDTO queryDTO) {
        return twitterEsService.countForExport(queryDTO);
    }

    @Override
    public List<TwitterUser> searchTwitterUsersWithEmail(String searchKeyword, int offset, int size) {
        List<TwitterUserEs> twitterUserEsList = twitterEsService.searchTwitterUsersWithEmail(searchKeyword, offset, size);

        List<TwitterUser> twitterUserList = new ArrayList<>();

        for (TwitterUserEs twitterUserEs : twitterUserEsList) {
            TwitterUser twitterUser = covertToTwitterUser(twitterUserEs);
            twitterUserList.add(twitterUser);
        }

        return twitterUserList;
    }

    @Override
    public Long countTwitterUsersWithEmail(String searchKeyword) {
        return twitterEsService.countTwitterUsersWithEmail(searchKeyword);

    }

    @Override
    public List<Map<String, Object>> searchByEmail(String value, int size, int offset) {
        List<Map<String, String>> maps = twitterEsService.searchByEmail(value, size, offset);

        List<Map<String, Object>> results = new ArrayList<>();
        for (Map<String, String> map : maps){
            String mysqlTable = map.get("sourceTable");
            String id = map.get("sourceId");
            Map<String, Object> twitterUserMap = getTwitterUserMapByTableAndId(mysqlTable, id);
            results.add(twitterUserMap);
        }
        return results;
    }
    
    /**
     * 获取Twitter用户的完整联系信息
     * @param sourceId 数据源ID
     * @param sourceTable 数据源表名
     * @param field 需要获取的字段名称
     * @return 返回包含完整字段信息的Map
     */
    @Override
    public Map<String, String> getFullContactInfo(String sourceId, String sourceTable, String field) {
        log.info("获取Twitter用户完整联系信息: sourceId={}, sourceTable={}, field={}", sourceId, sourceTable, field);
        Map<String, String> resultMap = new HashMap<>();
        
        try {
            // 参数验证
            if (!StringUtils.hasText(sourceId) || !StringUtils.hasText(sourceTable) || !StringUtils.hasText(field)) {
                log.warn("参数不能为空: sourceId={}, sourceTable={}, field={}", sourceId, sourceTable, field);
                return resultMap;
            }
            
            // 验证数据表名是否安全（防止SQL注入）
            if (!isValidTableName(sourceTable)) {
                log.warn("无效的数据表名: {}", sourceTable);
                return resultMap;
            }
            
            // 根据数据源信息查询用户数据
            List<Map<String, Object>> results = executeShardingSphereTwitterQuery(sourceTable, sourceId);
            
            if (results.isEmpty()) {
                log.warn("未找到Twitter用户数据: sourceId={}, sourceTable={}", sourceId, sourceTable);
                return resultMap;
            }
            
            // 获取查询结果
            Map<String, Object> userDataMap = results.get(0);
            
            // 获取需要的字段值
            String fieldValue = getString(userDataMap, field);
            if (fieldValue == null) {
                log.warn("未找到指定字段的值: field={}", field);
                return resultMap;
            }
            
            // 将字段值放入结果Map
            resultMap.put(field, fieldValue);
            log.info("成功获取Twitter用户完整联系信息: field={}, value={}", field, fieldValue);
            
            return resultMap;
        } catch (Exception e) {
            log.error("获取Twitter用户完整联系信息失败", e);
            return resultMap;
        }
    }

    @Override
    public List<TwitterUser> searchTwitterData(String companyName, String fullName,Integer matchType) {
        // 查询es,twitter只有邮箱
        List<TwitterUserEs> twitterUserEsList;

        twitterUserEsList =twitterEsService.searchTwitterDataWithEmail(companyName, fullName);

        if (twitterUserEsList.isEmpty()){
            return new ArrayList<>();
        }

        List<TwitterUser> twitterUserList = new ArrayList<>();

        for (TwitterUserEs twitterUserEs : twitterUserEsList) {
            TwitterUser twitterUser = covertToTwitterUser(twitterUserEs);
            twitterUserList.add(twitterUser);
        }

        return twitterUserList;

    }

    private Map<String, Object> getTwitterUserMapByTableAndId(String mysqlTable, String id) {

        log.info("根据ID查询企业: {}", id);
        try {
            log.info("根据数据表和ID查询企业: mysqlTable={}, id={}", mysqlTable, id);

            // 参数验证
            if (!StringUtils.hasText(mysqlTable) || !StringUtils.hasText(id)) {
                log.warn("参数不能为空: mysqlTable={}, id={}", mysqlTable, id);
                return null;
            }

            // 验证数据表名是否安全（防止SQL注入）
            if (!isValidTableName(mysqlTable)) {
                log.warn("无效的数据表名: {}", mysqlTable);
                return null;
            }
            // 查询twitter_user信息

            // 使用ShardingSphere兼容的查询方式
            List<Map<String, Object>> results = executeShardingSphereTwitterQuery(mysqlTable, id);

            if (results.isEmpty()) {
                log.info("未查询到企业信息: table={}, id={}", mysqlTable, id);
                return null;
            }

            // 将查询结果转换为TwitterUser对象
            return results.get(0);
        } catch (Exception e) {
            log.error("根据ID查询企业失败", e);
            return null;
        }
    }



    /**
     * 构建查询条件
     */
    private LambdaQueryWrapper<TwitterUser> buildQueryWrapper(TwitterQueryDTO queryDTO) {
        LambdaQueryWrapper<TwitterUser> queryWrapper = new LambdaQueryWrapper<>();
        
        // 名称搜索（name和screenName）
        if (StringUtils.hasText(queryDTO.getName())) {
            queryWrapper.and(wrapper -> 
                wrapper.like(TwitterUser::getName, queryDTO.getName())
                       .or()
                       .like(TwitterUser::getScreenName, queryDTO.getName())
            );
        }
        
        // 用户名搜索
        if (StringUtils.hasText(queryDTO.getScreenName())) {
            queryWrapper.like(TwitterUser::getScreenName, queryDTO.getScreenName());
        }
        
        // 邮箱搜索
        if (StringUtils.hasText(queryDTO.getEmail())) {
            queryWrapper.like(TwitterUser::getEmail, queryDTO.getEmail());
        }
        
        // 注册年份筛选
        if (queryDTO.getCreated_at_year() != null && !queryDTO.getCreated_at_year().isEmpty()) {
            queryWrapper.and(wrapper -> {
                for (int i = 0; i < queryDTO.getCreated_at_year().size(); i++) {
                    Integer year = queryDTO.getCreated_at_year().get(i);
                    // 拼接SQL，筛选指定年份的数据
                    String yearCondition = "YEAR(created_at) = " + year;
                    
                    if (i == 0) {
                        wrapper.apply(yearCondition);
                    } else {
                        wrapper.or().apply(yearCondition);
                    }
                }
            });
        }
        
        // 关注者数量范围筛选
        if (queryDTO.getFollowers_range() != null && queryDTO.getFollowers_range().length == 2) {
            int minFollowers = queryDTO.getFollowers_range()[0];
            int maxFollowers = queryDTO.getFollowers_range()[1];
            
            // 只要最小值大于等于0，最大值大于最小值，就应用筛选条件
            if (minFollowers >= 0 && maxFollowers > minFollowers) {
                queryWrapper.ge(TwitterUser::getFollowers, minFollowers)
                           .le(TwitterUser::getFollowers, maxFollowers);
            }
        }
        
        return queryWrapper;
    }
} 